package com.jianying.exer;

import com.jianying.beans.WaterSensor;
import com.jianying.func.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * ClassName: Flink_Kafka_Max_Mysql
 * Package: com.atguigu.exer
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/10 16:19
 * @Version: 1.0
 */
public class Flink_Kafka_Max_Mysql {
    public static void main(String[] args) throws Exception {
        // TODO 从Kafka读取WaterSensor数据,按照事件时间开窗(10秒滚动窗口),计算每个传感器的最高水位线,并将结果写出到MySQL。
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 读取数据 类型转换 指定水位线
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092,localhost:9093,localhost:9094")
                .setTopics("second")
                .setGroupId("test")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        SingleOutputStreamOperator<WaterSensor> kafkaSourceDS = env.fromSource(kafkaSource,
                        WatermarkStrategy
                                .<String>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                                    @Override
                                    public long extractTimestamp(String element, long recordTimestamp) {
                                        String[] words = element.split(",");
                                        return Long.valueOf(words[1]) * 1000;
                                    }
                                }),
                        "kafkaSource")
                .map(new WaterSensorMapFunction());

        // TODO 分组 开窗
        WindowedStream<WaterSensor, String, TimeWindow> windowDS = kafkaSourceDS.keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)));

        // TODO 数据处理
        SingleOutputStreamOperator<Tuple3<Long,String, Integer>> resultDS = windowDS.process(
                new ProcessWindowFunction<WaterSensor, Tuple3<Long,String, Integer>, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<Tuple3<Long,String, Integer>> out) throws Exception {
                        long windowEnd = context.window().getEnd();
                        int maxVC = 0;
                        for (WaterSensor element : elements) {
                            if (element.getVc() > maxVC) {
                                maxVC = element.getVc();
                            }
                        }
                        out.collect(Tuple3.of(windowEnd, s,maxVC));
                    }
                }
        );

        SinkFunction<Tuple3<Long, String, Integer>> mysqlSink = JdbcSink.<Tuple3<Long, String, Integer>>sink(
                "insert into test.maxVc values (?,?,?)",

                new JdbcStatementBuilder<Tuple3<Long, String, Integer>>() {
                    @Override
                    public void accept(PreparedStatement ps, Tuple3<Long, String, Integer> Tuple3Value) throws SQLException {
                        ps.setLong(1, Tuple3Value.f0);
                        ps.setString(2, Tuple3Value.f1);
                        ps.setInt(3, Tuple3Value.f2);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(5000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:13317/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8")
                        .withUsername("root")
                        .withPassword("123456")
                        .build()
        );

        resultDS.addSink(mysqlSink);


        env.execute();
    }
}
